package gupao.concurrency.datasync.task.v2;

import gupao.concurrency.datasync.entity.TciWordData;
import gupao.concurrency.datasync.task.AbstractSyncTask;
import gupao.concurrency.datasync.tbapi.Resp;
import gupao.concurrency.datasync.tbapi.TciWordDataApi;
import gupao.concurrency.utils.FutureUtil;
import gupao.concurrency.utils.JsonUtils;
import gupao.concurrency.utils.ThreadPoolService;
import gupao.concurrency.utils.Utils;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
@Component
public class SyncTaskV2 extends AbstractSyncTask {
    @Setter @Getter
    @RequiredArgsConstructor
    public static class SyncTaskStatusV2{
        private static final int ERROR_KEEP_MAX = 3;

        private final Date startTime;
        private final int providerConcurrency;
        private final int apiConcurrency;
        private final int dbConcurrency;

        private volatile Date endTime;

        private volatile long wordProvideTimeCost;
        private volatile long apiReaderTimeCost;
        private volatile long dbWriterTimeCost;
        private volatile long totalTimeCost;

        private final AtomicLong succNum = new AtomicLong(0);
        private final AtomicLong providerFail = new AtomicLong(0);
        private final AtomicLong apiFail = new AtomicLong(0);
        private final AtomicLong dbFail = new AtomicLong(0);
        private final String[] last3Error = new String[ERROR_KEEP_MAX];
        private final AtomicInteger errorIndex = new AtomicInteger(0);

        //单线程访问
        public void addError(String error) {
            int curIndex = errorIndex.get();
            int updateIndex = (curIndex + 1) % ERROR_KEEP_MAX;
            last3Error[curIndex] = error;
        }
    }
    @RequiredArgsConstructor
    public class WordProvider implements Callable<Integer> {
        final int num;
        private final int size;
        private int fail = 0;
        @Override
        public Integer call() throws Exception {
            log.info("WordProvider#{} is running by thread {}", num, Thread.currentThread().getName());
            for(int i = 0; i < size; i++){
                if(!wordQueue.offer(UUID.randomUUID().toString())){
                    fail++;
                }
                Utils.sleepIgnoreExp(1); //和v1一样模拟一个延迟
            }
            log.info("WordProvider#{} is finished by thread {}", num, Thread.currentThread().getName());
            return fail;
        }
    }
    @RequiredArgsConstructor
    public class ApiReader implements Callable<InnerTaskResultWrapper>{
        final int num;
        private int fail = 0;
        private int succ = 0;
        private String lastError;
        @Override
        public InnerTaskResultWrapper call() throws Exception {
            log.info("ApiReader#{} is running by thread {}", num, Thread.currentThread().getName());
            while (API_READER_RUNNING_FLAG){
                String word = wordQueue.poll();
                while(word != null){
                    Resp resp = api.syncTciWordReport(word);
                    if (resp.isSuccess() && resp.getData() != null
                            && tciWordDataQueue.offer(resp.getData())){
                        succ++;
                    } else {
                        fail++;
                        lastError = resp.getMsg();
                    }
                    word = wordQueue.poll();

                }
            }
            log.info("ApiReader#{} is finished by thread {}", num, Thread.currentThread().getName());
            return new InnerTaskResultWrapper(lastError,succ,fail);
        }
    }
    @RequiredArgsConstructor
    public class DbWriter implements Callable<InnerTaskResultWrapper>{
        final int num;
        private int fail = 0;
        private int succ = 0;
        private String lastError;
        private int batchSize = 20;
        @Override
        public InnerTaskResultWrapper call() throws Exception {
            log.info("DBWriter#{} is running by thread {}", num, Thread.currentThread().getName());
            List<TciWordData> batch = new ArrayList<>(batchSize);
            while(DB_WRITER_RUNNING_FLAG){
                TciWordData tciWord = tciWordDataQueue.poll();
                while(tciWord != null){
                    batch.add(tciWord);
                    if(batch.size() == batchSize){
                        batchInsert(batch);
                    }
                    tciWord = tciWordDataQueue.poll();
                }
            }
            if(batch.size() > 0){
                batchInsert(batch);
            }
            log.info("DBWriter#{} is finished by thread {}", num, Thread.currentThread().getName());
            return new InnerTaskResultWrapper(lastError,succ,fail);
        }

        private void batchInsert(List<TciWordData> batch){
            int x = tciWordMapper.insert(batch);
            if (x > 0) {
                succ += x;
            }
            if( x < batch.size() ){
                fail += batch.size() - x;
                lastError = "Insert DB Error";
            }
            batch.clear();
        }
    }
    private final ConcurrentLinkedQueue<String> wordQueue = new ConcurrentLinkedQueue<>();
    private final ConcurrentLinkedQueue<TciWordData> tciWordDataQueue = new ConcurrentLinkedQueue<>();

    private ThreadPoolExecutor providerService = ThreadPoolService.getOrCreate("provider",
            8,128, new SynchronousQueue<>(), new CallerRunsPolicy());

    private ThreadPoolExecutor apiReaderService = ThreadPoolService.getOrCreate("api-reader",
            8,512, new SynchronousQueue<>(), new CallerRunsPolicy());

    private ThreadPoolExecutor dbWriterService = ThreadPoolService.getOrCreate("db-writer",
            8,256,new SynchronousQueue<>(), new CallerRunsPolicy());

    private volatile boolean API_READER_RUNNING_FLAG = true;
    private volatile boolean DB_WRITER_RUNNING_FLAG = true;

    private final AtomicBoolean syncStatus = new AtomicBoolean(false);

    private volatile SyncTaskStatusV2 status;

    public TaskRunResultV2 getTaskResult(){
        return new TaskRunResultV2(providerService.getLargestPoolSize(),
                apiReaderService.getLargestPoolSize(),
                dbWriterService.getLargestPoolSize(),
                status);
    }

    @Autowired
    protected TciWordDataApi api;

    public void syncStart(int wordProviderNum, int apiReaderNum, int dbWriterNum, int totalNum){
        if(syncStatus.compareAndSet(false, true)){
            new Thread(() -> {
                try{
                    status = new SyncTaskStatusV2(new Date(),wordProviderNum,apiReaderNum,dbWriterNum);
                    API_READER_RUNNING_FLAG = true;
                    DB_WRITER_RUNNING_FLAG = true;

                    long wpStartTime = System.currentTimeMillis();
                    List<Future<Integer>> wpFutures = startProviders(wordProviderNum, totalNum);
                    long apiReaderStartTime = System.currentTimeMillis();
                    List<Future<InnerTaskResultWrapper>> apiReaderFutures = startApiReaders(apiReaderNum);
                    long dbWriterStartTime = System.currentTimeMillis();
                    List<Future<InnerTaskResultWrapper>> dbWriterFutures = startDbWriters(dbWriterNum);

                    List<Integer> wpFails = FutureUtil.waitFuturesComplete2(wpFutures);
                    long wpTimeCost = System.currentTimeMillis() - wpStartTime;
                    stopApiReaders(); // provider 结束，可以设置api readers 结束标识
                    status.setWordProvideTimeCost(wpTimeCost);

                    List<InnerTaskResultWrapper> apiResults = FutureUtil.waitFuturesComplete2(apiReaderFutures);
                    long apiTimeCost = System.currentTimeMillis() - apiReaderStartTime;
                    stopDbWriters();
                    status.setApiReaderTimeCost(apiTimeCost);

                    List<InnerTaskResultWrapper> dbResults = FutureUtil.waitFuturesComplete2(dbWriterFutures);
                    long dbTimeCost = System.currentTimeMillis() - dbWriterStartTime;
                    long totalTimeCost = System.currentTimeMillis() - wpStartTime;
                    status.setDbWriterTimeCost(dbTimeCost);
                    status.setTotalTimeCost(totalTimeCost);
                    status.setEndTime(new Date());
                    collectResultStatus(status,wpFails,apiResults,dbResults);
                    log.info("{}", JsonUtils.toJsonStr(getTaskResult()));
                }finally {
                    syncStatus.set(false);
                }
            }).start();
        }else{
            log.warn("Sync task is running.");
        }
    }

    private void collectResultStatus(SyncTaskStatusV2 status,
                                     List<Integer> wpFails,
                                     List<InnerTaskResultWrapper> apiResults,
                                     List<InnerTaskResultWrapper> dbResults){
        for(int f : wpFails){
            status.providerFail.addAndGet(f);
        }

        for(InnerTaskResultWrapper apiRt : apiResults){
            status.apiFail.addAndGet(apiRt.getFail());
            if(apiRt.getLastError() != null){
                status.addError(apiRt.getLastError());
            }
        }

        for(InnerTaskResultWrapper wbRt : dbResults){
            status.dbFail.addAndGet(wbRt.getFail());
            status.succNum.addAndGet(wbRt.getSucc());
            if(wbRt.getLastError() != null){
                status.addError(wbRt.getLastError());
            }
        }
    }

    private List<Future<Integer>> startProviders(int providerNum, int totalNum){
        List<Future<Integer>> fsList = new ArrayList<>(providerNum);
        int numPerProvider = totalNum / providerNum;
        int remains = totalNum % providerNum;
        for(int i = 0; i < providerNum - 1; i++){
            fsList.add(providerService.submit(new WordProvider(i,numPerProvider)));
        }
        fsList.add(providerService.submit(new WordProvider(providerNum - 1,numPerProvider + remains)));
        return fsList;
    }

    private List<Future<InnerTaskResultWrapper>> startApiReaders(int apiReaderNum){
        List<Future<InnerTaskResultWrapper>> fsList = new ArrayList<>(apiReaderNum);
        for(int i = 0; i < apiReaderNum; i++){
            fsList.add(apiReaderService.submit(new ApiReader(i)));
        }
        return fsList;
    }

    private List<Future<InnerTaskResultWrapper>> startDbWriters(int dbWriterNum){
        List<Future<InnerTaskResultWrapper>> fsList = new ArrayList<>(dbWriterNum);
        for(int i = 0; i < dbWriterNum; i++){
            fsList.add(dbWriterService.submit(new DbWriter(i)));
        }
        return fsList;
    }

    private void stopApiReaders(){
        API_READER_RUNNING_FLAG = false;
    }

    private void stopDbWriters(){
        DB_WRITER_RUNNING_FLAG = false;
    }

}
